package com.lucenten.mq;


import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQDestination;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.jms.*;


/**
 * 队列管理工具（1.队列的创建、2.队列销毁、3.消息发送、4.消息收取、5.关闭连接）
 * 
 * @author Azael
 *
 */
public class QueueMgrUtils {
	
	
	
	private final static Logger logger = LoggerFactory.getLogger(QueueMgrUtils.class);
	public static String mQbrokerURL ="";

	/**
	 * 创建生产者
	 */
	public static TSessionConnect createProducer(String queueName) throws Exception {
		TSessionConnect t = new TSessionConnect();
		// 实例化连接工厂
		ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("failover://"+mQbrokerURL);//tcp://20.13.3.23:61616
		Connection connection = null;
		Session session = null;
		MessageProducer messageProducer = null;
		try {
			// 通过连接工厂获取连接
			connection = connectionFactory.createConnection();
			// 启动连接
			connection.start();
			// 创建session
			session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
			// 创建队列
			Destination destination = session.createQueue(queueName);
			// 创建消息生产者
			messageProducer = session.createProducer(destination);
			// 设置生成者对象
			t.setConnection(connection);
			t.setMessageProducer(messageProducer);
			t.setSession(session);
		} catch (Exception e) {
			e.printStackTrace();
			closeConnSession(connection, session);
			logger.error("创建队列生成者失败！" + e.getMessage());
			throw new Exception("创建队列生成者失败！" + e.getMessage());
		}
		return t;
	}

	/**
	 * 创建消费者
	 */
	public static TSessionConnect createConsumer(String queueName) throws Exception {
		TSessionConnect t = new TSessionConnect();
		// 实例化连接工厂
		ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("failover://"+mQbrokerURL);
		Connection connection = null;
		Session session = null;
		try {
			// 通过连接工厂获取连接
			connection = connectionFactory.createConnection();
			// 启动连接
			connection.start();
			// 创建session
			session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
			// 创建队列
			Destination destination = session.createQueue(queueName);
			// 创建消息消费者
			MessageConsumer messageConsumer = session.createConsumer(destination);
			// 设置消费者对象
			t.setConnection(connection);
			t.setMessageConsumer(messageConsumer);
			t.setSession(session);
		} catch (Exception e) {
			e.printStackTrace();
			closeConnSession(connection, session);
			logger.error("创建队列消费者失败！" + e.getMessage());
			throw new Exception("创建队列消费者失败！" + e.getMessage());
		}
		return t;
	}

	/**
	 * 通过生产者发送消息
	 * 
	 * @param sessionConn
	 * @param message
	 * @return
	 */
	public static boolean sendMessageToQueue(TSessionConnect sessionConn, String message) throws Exception {
		boolean flag = true;
		try {
			// 发送消息
			TextMessage textmessage = sessionConn.getSession().createTextMessage(message);
			sessionConn.getMessageProducer().send(textmessage);
		} catch (JMSException e) {
			e.printStackTrace();
			flag = false;
			logger.error("发送消息失败！" + e.getMessage());
			throw new Exception("发送消息失败！" + e.getMessage());
		}
		return flag;
	}

	/**
	 * 销毁队列
	 */
	public static void destoryQueue(String queueName) {
		// 实例化连接工厂
		ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
		ActiveMQConnection connection = null;
		Session session = null;
		try {
			// 通过连接工厂获取连接
			connection = (ActiveMQConnection) connectionFactory.createConnection();
			// 启动连接
			connection.start();
			// 创建session
			session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
			// 创建队列
			ActiveMQDestination destination = (ActiveMQDestination) session.createQueue(queueName);
			// 销毁队列
			connection.destroyDestination(destination);
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			closeConnSession(connection, session);
		}
	}

	/**
	 * 关闭会话和连接
	 * 
	 * @param connection
	 * @param session
	 */
	public static void closeConnSession(Connection connection, Session session) {
		if (session != null) {
			try {
				session.close();
			} catch (JMSException e) {
				e.printStackTrace();
			}
		}
		if (connection != null) {
			try {
				connection.close();
			} catch (JMSException e) {
				e.printStackTrace();
			}
		}
	}

}